package com.nncsys.datastream

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * Created with IntelliJ IDEA
 * <p>说明：</p>
 *
 * @author micheal
 * @see
 * @since 2020/04/01 0001 15:13
 */
object DataStreamWcDemo {

  var hostname = "localhost"
  var port = 9999
  def parse(list: List[String]): Unit = list match {
    case "--hostname" :: value :: tail =>
      hostname = value;
      parse(tail)

    case "--port" :: value :: tail =>
      port = value.toInt
      parse(tail)
    case Nil =>

  }


  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
    parse(args.toList)
    println(hostname,port)

    val data = env.socketTextStream(hostname,port)

    import org.apache.flink.api.scala._
    val result = data
      .flatMap(_.toLowerCase.split(" "))
      .filter(_.nonEmpty)
      .map((_, 1))
      .keyBy(0)
      .timeWindow( Time.seconds((2)) )
      .sum(1)

    result.print().setParallelism(1)

    println("start...")
    env.execute("stream word count")

  }

}
